package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;

public class Demo17_ChangelogTableToStream {

    public static void main(String[] args) throws Exception {

        // 构建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 用连接器建表，映射物理数据源
        tenv.executeSql(
                " create table stu_info_cdc(         " +
                        "    id int,                         " +
                        "    name string,                    " +
                        "    gender string,                  " +
                        "    age int,                        " +
                        "    score float,                    " +
                        "    ts bigint,                        " +
                        "    rt as to_timestamp_ltz(ts,3),   " +
                        "    watermark for rt as rt,         " +
                        "    primary key(id) not enforced    " +
                        " ) WITH (                           " +
                        "     'connector' = 'mysql-cdc',     " +
                        "     'hostname' = 'doitedu',        " +
                        "     'port' = '3306',               " +
                        "     'username' = 'root',           " +
                        "     'password' = 'root',           " +
                        "     'database-name' = 'doit47',    " +
                        "     'table-name' = 'stu_info'      " +
                        " )                                  "
        );


        Table table = tenv.from("stu_info_cdc");

        // 如果表中有update和delete的行类型，则不能用如下api去转流
        //DataStream<Row> dataStream = tenv.toDataStream(table);

        // 表转流时，watermark会自动延续
        DataStream<Row> changelogStream = tenv.toChangelogStream(table);

        changelogStream.process(new ProcessFunction<Row, String>() {
            @Override
            public void processElement(Row row, ProcessFunction<Row, String>.Context ctx, Collector<String> out) throws Exception {

                // row中有行类型，叫做rowKind
                RowKind kind = row.getKind();

                //if(kind.shortString().equals("+I") )
                if (!(kind == RowKind.DELETE)) {
                    // 别的字段
                    int id = row.getFieldAs("id");
                    String name = row.getFieldAs("name");


                    // 输出成csv拼接的字符串
                    out.collect(kind.shortString() + "," + id + "," + name + "," + ctx.timerService().currentWatermark());

                }

            }
        }).print();


        env.execute();
    }

}
